/*
 * Copyright (C) 2014 The Guava Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
 * in compliance with the License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package org.jsets.fastboot.eventbus;

import static com.google.common.base.Preconditions.checkNotNull;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.eventbus.AllowConcurrentEvents;
import com.google.j2objc.annotations.Weak;

import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Objects;
import java.util.concurrent.Executor;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.support.StandardEvaluationContext;

/**
 * 
 * copy from guava-30.0-jre
 * 
 * A subscriber method on a specific object, plus the executor that should be used for dispatching
 * events to it.
 *
 * <p>Two subscribers are equivalent when they refer to the same method on the same object (not
 * class). This property is used to ensure that no subscriber method is registered more than once.
 *
 * @author Colin Decker
 */
@Slf4j
class Subscriber {

  /** Creates a {@code Subscriber} for {@code method} on {@code listener}. */
  static Subscriber create(DefaultEventBus bus, Object listener, Method method) {
    return isDeclaredThreadSafe(method)
        ? new Subscriber(bus, listener, method)
        : new SynchronizedSubscriber(bus, listener, method);
  }

  /** The event bus this subscriber belongs to. */
  @Weak private DefaultEventBus bus;

  /** The object with the subscriber method. */
  @VisibleForTesting final Object target;

  /** Subscriber method. */
  private final Method method;

  /** Executor to use for dispatching events to this subscriber. */
  private final Executor executor;
  
  /** 条件表达式 */
  private Expression conditionExpression;
  
  /** Subscriber名称 */
  private String key;
  /** 参数名称 */
  private String parameterName;
  /** 是否异步 */
  private boolean async;
  /** 是否重试 */
  private boolean retryAvailable;
  private Class<? extends Throwable>[] retryFor;
  /** 重试次数，默认3*/
  private int maxAttempts;
  /**每次重试延迟毫秒数，默认为0*/
  private int delay;
  /** 延迟倍数）默认为0；大于0时生效；如果delay等于2，multiplier等于2，则第一次重试为2秒，第二次为4秒，第三次为8秒*/
  private int multiplier;

  private Subscriber(DefaultEventBus bus, Object target, Method method) {
    this.bus = bus;
    this.target = checkNotNull(target);
    this.method = method;
    method.setAccessible(true);

    this.executor = bus.executor();
    
    if (this.bus instanceof AsyncEventBus) {
		this.async = true;
	}
  }

  /** Dispatches {@code event} to this subscriber using the proper executor. */
  final void dispatchEvent(final Object event) {
    executor.execute(
        new Runnable() {
          @Override
          public void run() {
        	  if(retryAvailable) {
        		 if(async) { //异步
           		 	try {
           		 		invokeSubscriberMethod(event);
           		 		log.info("事件处理成功，事件["+event.toString()+"]，监听器["+target.toString()+"]");
           		 	} catch (InvocationTargetException e) {
                  	  if(retryNecessary(e)) {
                  		  RetryTask retryTask = new RetryTask(event, key, delay, 1);
                  		  RetryQueue.getInstance().put(retryTask);
                  	  }else {
                  		  bus.handleSubscriberException(e.getCause(), context(event));
                  	  }
           		 	}
        		 }else { //同步
        			int exponent = delay;
           		  	int index = 0;
           		  	while(index<=maxAttempts) {
           		  		try {
           		  			if(index>0) {
                       			log.info("第"+index+"次重试，事件["+event.toString()+"]，监听器["+target.toString()+"]");
           		  			}
           		  			invokeSubscriberMethod(event);
           		  			log.info("事件处理成功，事件["+event.toString()+"]，监听器["+target.toString()+"]");
           		  			break;
                         } catch (InvocationTargetException e) {
                       	  	boolean necessary = retryNecessary(e);
                       	  	if(!necessary) {
                       	  		index=maxAttempts;
                       	  	}
                       	  	
                       	  	if(index == maxAttempts) {
                    	  		bus.handleSubscriberException(e.getCause(), context(event));
                    	  	} else {
                    	  		if(exponent>0) {
                    	  			try {
                    	  				Thread.sleep(exponent*1000);
                    	  			} catch (InterruptedException e1) {
                    	  				// 忽略...
                    	  			}
									exponent = exponent * multiplier;
                    	  		}
                    	  	}
                         } finally {
                       	  	index++;
                         }
           		  	}
        		 }
        		 //
        	  } else { // 不需要重试
        		  try {
                      invokeSubscriberMethod(event);
                      log.info("事件处理成功，事件["+event.toString()+"]，监听器["+target.toString()+"]");
                  } catch (InvocationTargetException e) {
                      bus.handleSubscriberException(e.getCause(), context(event));
                  }
        	  }
          }
        });
  }

  final boolean dispatchNecessary(final Object event) {
	if(Objects.nonNull(this.conditionExpression)) {
		EvaluationContext context = new StandardEvaluationContext();
		context.setVariable(this.parameterName, event);
		Boolean necessary = this.conditionExpression.getValue(context,Boolean.class);
		if(Objects.nonNull(necessary)&&necessary) {
			return true;
		}
		return false;
	}
	return true;
  }
  
  /**
   * Invokes the subscriber method. This method can be overridden to make the invocation
   * synchronized.
   */
  @VisibleForTesting
  void invokeSubscriberMethod(Object event) throws InvocationTargetException {
    try {
      method.invoke(target, checkNotNull(event));
    } catch (IllegalArgumentException e) {
      throw new Error("Method rejected target/argument: " + event, e);
    } catch (IllegalAccessException e) {
      throw new Error("Method became inaccessible: " + event, e);
    } catch (InvocationTargetException e) {
      if (e.getCause() instanceof Error) {
        throw (Error) e.getCause();
      }
      throw e;
    }
  }

  /** Gets the context for the given event. */
  private SubscriberExceptionContext context(Object event) {
    return new SubscriberExceptionContext(bus, event, target, method);
  }

  @Override
  public final int hashCode() {
    return (31 + method.hashCode()) * 31 + System.identityHashCode(target);
  }

  @Override
  public final boolean equals(@Nullable Object obj) {
    if (obj instanceof Subscriber) {
      Subscriber that = (Subscriber) obj;
      // Use == so that different equal instances will still receive events.
      // We only guard against the case that the same object is registered
      // multiple times
      return target == that.target && method.equals(that.method);
    }
    return false;
  }

  /**
   * Checks whether {@code method} is thread-safe, as indicated by the presence of the {@link
   * AllowConcurrentEvents} annotation.
   */
  private static boolean isDeclaredThreadSafe(Method method) {
    return method.getAnnotation(AllowConcurrentEvents.class) != null;
  }
  
  /**
   * 设置条件表达式
   * @param conditionExpression 
   */
  public void setConditionExpression(Expression conditionExpression) {
	this.conditionExpression = conditionExpression;
  }
  
  /**
   * 设置参数名称
   * @param parameterName 
   */
  public void setParameterName(String parameterName) {
	this.parameterName = parameterName;
  }
  
  /**
   * 设置是否重试
   * @param retryAvailable 
   */
  public void setRetryAvailable(boolean retryAvailable) {
	this.retryAvailable = retryAvailable;
  }

  public void setRetryFor(Class<? extends Throwable>[] retryFor) {
	this.retryFor = retryFor;
  }

  /**
   * 设置重试次数
   * @param maxAttempts 
   */
  public void setMaxAttempts(int maxAttempts) {
	  this.maxAttempts = maxAttempts;
  }

  /**
   * 设置每次重试延迟毫秒数
   * @param delay 
   */
  public void setDelay(int delay) {
	  this.delay = delay;
  }

  public int getDelay() {
	  return delay;
  }

  public int getMultiplier() {
	  return multiplier;
  }

  public DefaultEventBus getBus() {
	return bus;
  }

  /**
   * 设置延迟倍数
   * @param multiplier 
   */
  public void setMultiplier(int multiplier) {
	this.multiplier = multiplier;
  }

  public int getMaxAttempts() {
	return maxAttempts;
  }

  public void setKey(String key) {
	this.key = key;
  }
  
  public Object getTarget() {
	return target;
  }

  public Method getMethod() {
	return method;
  }

  public boolean retryNecessary(Exception e) {
	  if(this.retryFor==null) {
		  return true;
	  }
	  
	  if(0==this.retryFor.length) {
		  return true;
	  }
	  
	  Throwable cause = e.getCause();
	  if(null == cause) {
		  return true;
	  }
	  
	  for(Class<? extends Throwable> clazz : this.retryFor) {
		  if(clazz.equals(cause.getClass())) {
			  return true;
		  }
	  }
	  
	  return false;
  }
  
  /**
   * Subscriber that synchronizes invocations of a method to ensure that only one thread may enter
   * the method at a time.
   */
  @VisibleForTesting
  static final class SynchronizedSubscriber extends Subscriber {

    private SynchronizedSubscriber(DefaultEventBus bus, Object target, Method method) {
      super(bus, target, method);
    }

    @Override
    void invokeSubscriberMethod(Object event) throws InvocationTargetException {
      synchronized (this) {
        super.invokeSubscriberMethod(event);
      }
    }
  }
}
